package com.seaboxdata.jt809.kafka;

import com.seaboxdata.jt809.common.util.PropertiesUtil;
import com.seaboxdata.jt809.db.JT809Dao;
import com.seaboxdata.jt809.kafka.serializer.JT809Packet0x1202Deserializer;
import com.seaboxdata.jt809.kafka.serializer.JT809Packet0x1202Serializer;
import com.seaboxdata.jt809.kafka.serializer.JT809Packet0x1402Serializer;
import com.seaboxdata.jt809.packet.JT809Packet0x1202;
import com.seaboxdata.jt809.packet.JT809Packet0x1402;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * @ClassName JT809Packet0x1402MessageProducer
 * @Description TODO
 * @Author DFJX
 * @Company 北京东方金信科技有限公司
 * @Date 2020/7/21 13:55
 * Version 1.0
 **/
public class JT809Packet0x1402MessageProducer {
    Producer<String,JT809Packet0x1402> producer;
    public JT809Packet0x1402MessageProducer(){
        Properties props = new Properties();
        // 服务器ip
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, PropertiesUtil.getProperty("kafka.server"));
        // 属性键值对都序列化成字符串
        // props.put("")
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,1000);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JT809Packet0x1402Serializer.class.getName());
        producer = new KafkaProducer<String,JT809Packet0x1402>(props);
    }
    public void sendMessageToKafka(JT809Packet0x1402 obj){
        producer.send(new ProducerRecord<String,JT809Packet0x1402>("warn",obj));
       // producer.close();
    }
}
